package com.lvmama.rhino.analyze.client

import com.google.common.base.CharMatcher
import com.lvmama.rhino.common.entity.{JobConfig, MonitorPageLog}
import com.lvmama.rhino.common.utils.Utils
import com.lvmama.rhino.common.utils.logging.LazyLogging
import com.lvmama.rhino.common.utils.spark.SparkApplication
import com.lvmama.rhino.common.utils.JDBCUtil.Conversions.ImplicitInsert
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.StringType
import org.apache.spark.storage.StorageLevel


/**
  * Created by Administrator on 2017/7/18.
  */
class MonitorPage(config: JobConfig, path: String) extends SparkApplication with LazyLogging{
  override var appName: String = "MonitorPageJob"
  override var sparkConfig: Map[String, String] = config.spark

  def execute(): Unit = {
    sparkConfig += ("spark.app.name" -> appName)
    sparkConfig += ("spark.master" -> "local[3]")
    withSparkContext { sc =>
      val sqlCtx = new SQLContext(sc)
      import sqlCtx.implicits._
      val log = sc.textFile(path).map(l => Utils.initMonitorPage(l)).toDF()
      val result = log.filter(!col("createDate").equalTo(""))
      val monitorPageData = result.withColumn("ext_url",processUrl(col("url").cast(StringType)))
        .withColumn("whiteScreenInterval",when(col("whiteScreenTime").between(0,1),0).when(col("whiteScreenTime").between(1,2),1)
          .when(col("whiteScreenTime").between(2,3),2).when(col("whiteScreenTime").between(3,4),3).when(col("whiteScreenTime").between(4,5),4)
          .when(col("whiteScreenTime").>(5),5))
        .withColumn("firstScreenInterval",when(col("firstScreenTime").between(0,1),0).when(col("firstScreenTime").between(1,2),1)
          .when(col("firstScreenTime").between(2,3),2).when(col("firstScreenTime").between(3,4),3).when(col("firstScreenTime").between(4,5),4)
          .when(col("firstScreenTime").>(5),5))
        .withColumn("interactiveInterval",when(col("interactiveTime").between(0,1),0).when(col("interactiveTime").between(1,2),1)
          .when(col("interactiveTime").between(2,3),2).when(col("interactiveTime").between(3,4),3).when(col("interactiveTime").between(4,5),4)
          .when(col("interactiveTime").>(5),5))
        .withColumn("pageLoadInterval",when(col("pageLoadTime").between(0,2),0).when(col("pageLoadTime").between(2,4),1)
          .when(col("pageLoadTime").between(4,6),2).when(col("pageLoadTime").between(6,8),3).when(col("pageLoadTime").between(8,10),4)
          .when(col("pageLoadTime").>(10),5))
        .persist(StorageLevel.MEMORY_AND_DISK_SER)

      val columns = Seq("ext_url", "platform", "createDate", "whiteScreenTime", "firstScreenTime", "interactiveTime", "downloadTime"
                      ,"dnsTime", "buildUnionTime", "sslTime", "pageLoadTime", "backendProcessTime")
      val contactColumns = Seq("ext_url", "platform", "createDate", "whiteScreenTime", "firstScreenTime", "interactiveTime", "downloadTime"
                      ,"dnsTime", "buildUnionTime", "sslTime", "pageLoadTime", "backendProcessTime", "concatCol")

      val data = monitorPageData.select(columns.map(col): _*)
        .withColumn("concatCol",concat(col("ext_url"), col("platform"), col("createDate")))
        .select(contactColumns.map(col): _*)
        .groupBy("ext_url","platform","createDate","concatCol")
        .agg(avg(col("whiteScreenTime")).as("whiteScreenTime"),avg(col("firstScreenTime")).as("firstScreenTime"),
          avg(col("interactiveTime")).as("interactiveTime"),avg(col("downloadTime")).as("downloadTime"),
          avg(col("dnsTime")).as("dnsTime"),avg(col("buildUnionTime")).as("buildUnionTime"),
          avg(col("pageLoadTime")).as("pageLoadTime"),
          avg(col("backendProcessTime")).as("backendProcessTime"),count("*").as("amount"))

      val successMonitorPageData = monitorPageData.filter(col("sslFlag").equalTo("1"))
      val successData = successMonitorPageData.select(columns.map(col): _*)
        .withColumn("concatCol",concat(col("ext_url"), col("platform"), col("createDate")))
        .select(contactColumns.map(col): _*)
        .groupBy("ext_url","platform","createDate","concatCol")
        .agg(avg(col("sslTime")).as("sslTime"),count("*").as("sslAmount"))
      //数据整合
      val resultData = data.join(successData, data.col("concatCol")===successData.col("concatCol"), "left_outer").select(data.col("*"), successData.col("sslTime"), successData.col("sslAmount"))
        .na.fill(0, Seq("sslAmount"))

      resultData.select(col("ext_url").as("url"),col("platform"),col("whiteScreenTime").as("white_screen_time"),
        col("firstScreenTime").as("first_screen_time"),col("interactiveTime").as("interactive_time"),col("downloadTime").as("download_time"),
        col("dnsTime").as("dns_time"),col("buildUnionTime").as("build_union_time"),col("sslTime").as("ssl_time"),
        col("pageLoadTime").as("page_load_time"),col("backendProcessTime").as("backend_process_time"),col("amount"),col("createDate").as("create_date"),col("sslAmount"))
        .insertDF2MysqlToWolverine("monitor_page")

      val whiteScreenData = monitorPageData.select(col("ext_url"),col("whiteScreenInterval"),col("createDate"))
        .groupBy("ext_url","whiteScreenInterval","createDate")
        .agg(count("*").as("amount"))

      val firstScreenData = monitorPageData.select(col("ext_url"),col("firstScreenInterval"),col("createDate"))
        .groupBy("ext_url","firstScreenInterval","createDate")
        .agg(count("*").as("amount"))

      val interactiveData = monitorPageData.select(col("ext_url"),col("interactiveInterval"),col("createDate"))
        .groupBy("ext_url","interactiveInterval","createDate")
        .agg(count("*").as("amount"))

      val pageLoadData = monitorPageData.select(col("ext_url"),col("pageLoadInterval"),col("createDate"))
        .groupBy("ext_url","pageLoadInterval","createDate")
        .agg(count("*").as("amount"))

      whiteScreenData.select(col("ext_url").as("url"),col("whiteScreenInterval").as("time_interval"),col("createDate").as("create_date"),col("amount"))
       .insertDF2MysqlToWolverine("white_screen_time")

      firstScreenData.select(col("ext_url").as("url"),col("firstScreenInterval").as("time_interval"),col("createDate").as("create_date"),col("amount"))
        .insertDF2MysqlToWolverine("first_screen_time")

      interactiveData.select(col("ext_url").as("url"),col("interactiveInterval").as("time_interval"),col("createDate").as("create_date"),col("amount"))
        .insertDF2MysqlToWolverine("interactive_time")

      pageLoadData.select(col("ext_url").as("url"),col("pageLoadInterval").as("time_interval"),col("createDate").as("create_date"),col("amount"))
        .insertDF2MysqlToWolverine("page_load_time")

      monitorPageData.unpersist()
    }
  }

  def processUrl = udf((url: String) => {
    var ext_url = ""
    if(url.contains("zt1.lvmama.com/")){
      val last = url.lastIndexOf("/")
      val prefixes = url.substring(0,last)
      val suffix = url.substring(last,url.length())
      var ext_suffix = CharMatcher.JAVA_DIGIT.replaceFrom(suffix, "*")
      ext_suffix = ext_suffix.replaceAll("[*]+", "*")
      ext_url = prefixes + ext_suffix
    }else if(url.contains("static/zt/")){
      val first = url.indexOf("static/zt/")+10
      val suffix0 = url.substring(first,url.length())
      val suffix_first= suffix0.indexOf("/")
      val index = first + suffix_first
      val prefixes = url.substring(0,index)
      val suffix = url.substring(index,url.length())
      var ext_suffix = CharMatcher.JAVA_DIGIT.replaceFrom(suffix, "*")
      ext_suffix = ext_suffix.replaceAll("[*]+", "*")
      ext_url = prefixes + ext_suffix
    }else{
      ext_url = CharMatcher.JAVA_DIGIT.replaceFrom(url, "*")
      ext_url = ext_url.replaceAll("[*]+", "*")
    }
    ext_url
  })

}

object MonitorPage{
  def apply(path: String): Unit = {
    val config = JobConfig()
    new MonitorPage(config, path).execute()
  }
}

